package org.example.customer;

import com.alibaba.fastjson.JSONArray;
import jakarta.annotation.Resource;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
import org.example.bean.Alipay;
import org.example.bean.Logistics;
import org.example.mapper.LogisticsMapper;
import org.example.service.LogisticsService;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;

/**
 * @Auther:Liu
 * @Testname:LogisticsCustomer
 * @Date:2025/8/23 17:21
 */
@Slf4j
@Component
public class LogisticsCustomer {

    @Resource
    private LogisticsService logisticsService;

    @SneakyThrows
    @Scheduled(initialDelay = 1000)
    public void a() {
        log.info("开启物流");
        log.info("<logisticsService:{}>",logisticsService);
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("producer_group_c");
        consumer.setNamesrvAddr("192.168.10.88:9876");
        consumer.setMessageModel(MessageModel.BROADCASTING);
        consumer.subscribe("Order", "alipay");
        consumer.registerMessageListener((MessageListenerConcurrently) (list, consumeConcurrentlyContext) -> {
            log.info("物流收到信息");
            for (MessageExt msg : list) {
                byte[] body = msg.getBody();
                String bodyStr = new String(body, StandardCharsets.UTF_8);
                Alipay alipay = JSONArray.parseObject(bodyStr, Alipay.class);
                //添加物流
                Logistics logistics = new Logistics();
                logistics.setName(alipay.getName());
                logistics.setAddress(alipay.getAddress());
                logistics.setPhone(alipay.getPhone());
                logisticsService.save(logistics);
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
    }
}
